package org.example.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class HBaseUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseUtil.class);
    private static Configuration conf;
    private static Connection connection;

    public static final TableName TABLE_NAME = TableName.valueOf("database");
    public static final byte[] CF_INFO = Bytes.toBytes("info");
    public static final byte[] CF_VEHICLE = Bytes.toBytes("vehicle");
    public static final byte[] CF_RECALL = Bytes.toBytes("recall");

    static {
        conf = HBaseConfiguration.create();
        
        // 根据你的集群配置进行设置
        // ZooKeeper集群配置 - 临时排除有问题的node02
        conf.set("hbase.zookeeper.quorum", "node01,node03");  // 临时排除node02
        // conf.set("hbase.zookeeper.quorum", "node01,node02,node03");  // 原始配置
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        
        // HBase集群配置
        conf.set("hbase.cluster.distributed", "true");
        conf.set("hbase.master.port", "16000");
        
        // HDFS配置 - 这是关键！
        conf.set("fs.defaultFS", "hdfs://node01:8020");
        conf.set("hbase.rootdir", "hdfs://node01:8020/hbase");
        
        // 客户端连接配置 - 增加重试次数和超时时间
        conf.set("hbase.client.retries.number", "10");  // 增加重试次数
        conf.set("hbase.client.pause", "200");  // 增加暂停时间
        conf.set("hbase.rpc.timeout", "60000");  // 增加RPC超时
        conf.set("hbase.client.operation.timeout", "120000");  // 增加操作超时
        conf.set("hbase.client.scanner.timeout.period", "300000");  // 增加扫描超时
        
        // 分布式环境的安全设置
        conf.set("hbase.unsafe.stream.capability.enforce", "false");
        
        // 网络和连接相关配置
        conf.set("hbase.client.ipc.pool.type", "RoundRobinPool");
        conf.set("hbase.client.ipc.pool.size", "1");
        conf.set("hbase.regionserver.handler.count", "30");
        
        // 禁用Kerberos认证（如果启用的话）
        conf.set("hbase.security.authentication", "simple");
        conf.set("hbase.security.authorization", "false");
        
        // 添加容错配置
        conf.set("hbase.client.fail.fast", "false");  // 不快速失败
        conf.set("hbase.client.retry.policy", "org.apache.hadoop.hbase.client.ExponentialBackoffRetry");
        
        try {
            LOGGER.info("正在尝试连接到HBase集群...");
            LOGGER.info("ZooKeeper配置: {}", conf.get("hbase.zookeeper.quorum"));
            LOGGER.info("HBase Master端口: {}", conf.get("hbase.master.port"));
            LOGGER.info("HDFS地址: {}", conf.get("fs.defaultFS"));
            connection = ConnectionFactory.createConnection(conf);
            LOGGER.info("HBase连接成功！");
        } catch (IOException e) {
            LOGGER.error("HBase连接失败！请检查Zookeeper和HBase Master是否正在运行，以及配置是否正确。", e);
            LOGGER.error("当前ZooKeeper配置: {}", conf.get("hbase.zookeeper.quorum"));
            LOGGER.error("当前HDFS配置: {}", conf.get("fs.defaultFS"));
        }
    }

    public static Connection getConnection() {
        return connection;
    }

    public static void createTable() throws IOException {
        try (Admin admin = connection.getAdmin()) {
            if (admin.tableExists(TABLE_NAME)) {
                LOGGER.warn("表 '{}' 已存在，无需创建。", TABLE_NAME);
            } else {
                LOGGER.info("表 '{}' 不存在，现在开始创建...", TABLE_NAME);
                TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TABLE_NAME)
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF_INFO))
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF_VEHICLE))
                        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(CF_RECALL))
                        .build();
                admin.createTable(tableDescriptor);
                LOGGER.info("表 '{}' 创建成功！", TABLE_NAME);
            }
        }
    }

    public static void recreateTable() throws IOException {
        try (Admin admin = connection.getAdmin()) {
            if (admin.tableExists(TABLE_NAME)) {
                LOGGER.warn("表 '{}' 已存在，将进行清空重建操作...", TABLE_NAME);
                admin.disableTable(TABLE_NAME);
                admin.deleteTable(TABLE_NAME);
                LOGGER.info("旧表 '{}' 已成功删除。", TABLE_NAME);
            }
            // 调用现有的建表方法来创建新表
            createTable();
        }
    }

    public static void close() throws IOException {
        if (connection != null && !connection.isClosed()) {
            LOGGER.info("正在关闭HBase连接...");
            connection.close();
            LOGGER.info("HBase连接已关闭。");
        }
    }
} 